package com.zengjianjun.canal.demo.ordinary;

import cn.hutool.json.JSONUtil;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Component;

import java.net.InetSocketAddress;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;
import java.util.stream.Collectors;

/**
 * 普通消费
 *
 * @author zengjianjun
 */
@Slf4j
@Component
public class CanalTestConsumer implements CommandLineRunner {

    @Value("${canal.hostname:192.168.137.128}")
    private String canalHostname;
    @Value("${canal.port:11111}")
    private int canalPort;

    @Override
    public void run(String... args) throws Exception {
//        new Thread(this::start).start();
    }

    public void start() {
        final String canalSubscribe = "py_web.test";
        final String canalDestination = "test";
        log.info("canalHostname = {}, canalPort = {}, canal.subscribe = {}, canal.destination = {}", canalHostname, canalPort, canalSubscribe, canalDestination);
        // 创建连接，配置连接参数
        CanalConnector connector = CanalConnectors.newSingleConnector(
                new InetSocketAddress(canalHostname, canalPort),
                canalDestination,
                "",
                ""
        );
        connector.connect();
        // 设置订阅规则
        connector.subscribe(canalSubscribe);
        // 从头开始消费
        connector.rollback();
        while (true) {
            try {
                Message message = connector.getWithoutAck(1000, 1L, TimeUnit.SECONDS);
                long batchId = message.getId();
                if (batchId != -1 && !message.getEntries().isEmpty()) {
                    log.info("batchId = {}, size = {}", batchId, message.getEntries().size());
                    messageProcess(message);
                    connector.ack(batchId);
                }
                Thread.sleep(50);
            } catch (Exception e) {
                log.error("canal get message error: ", e);
            }
        }
    }

    /**
     * 消息处理
     *
     * @param message 消息数据
     */
    private void messageProcess(Message message) {
        try {
            // 数据过滤
            List<CanalEntry.Entry> list = message.getEntries().stream().filter(data -> {
                try {
                    CanalEntry.EntryType entryType = data.getEntryType();
                    if (entryType == CanalEntry.EntryType.TRANSACTIONBEGIN || entryType == CanalEntry.EntryType.TRANSACTIONEND) {
                        // 过滤开启/关闭事务操作类型
                        return false;
                    }
                    CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(data.getStoreValue());
                    // 获取操作类型：insert/update/delete类型
                    CanalEntry.EventType eventType = rowChange.getEventType();
                    if (eventType != CanalEntry.EventType.INSERT) {
                        // 非新增类型
                        return false;
                    }
                } catch (Exception e) {
                    log.error("message filter error: data = {}", JSONUtil.toJsonStr(data), e);
                    return false;
                }
                return true;
            }).collect(Collectors.toList());
            // 数据解析
            int size = list.size();
            int batchSize = 0;
            for (int i = 0; i < size; i++) {
                CanalEntry.Entry entry = list.get(i);
                // 打印当前批次最后一条Header信息
                if (i == size - 1) {
                    log.info(String.format("binlog[%s:%s] , dbName:%s, tableName:%s ",
                            entry.getHeader().getLogfileName(), entry.getHeader().getLogfileOffset(),
                            entry.getHeader().getSchemaName(), entry.getHeader().getTableName()));
                }
                CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
                List<CanalEntry.RowData> rowDataList = rowChange.getRowDatasList();
                batchSize += rowDataList.size();
                for (int j = 0; j < rowDataList.size(); j++) {
                    CanalEntry.RowData rowData = rowDataList.get(j);
                    List<CanalEntry.Column> columnList = rowData.getAfterColumnsList();
                    Map<String, CanalEntry.Column> columnMap = columnList.stream().collect(Collectors.toMap(CanalEntry.Column::getName, Function.identity()));
                    CanalEntry.Column idColum = columnMap.get("id");
                    CanalEntry.Column usernameColumn = columnMap.get("username");
                    CanalEntry.Column passwordColumn = columnMap.get("password");
                    CanalEntry.Column createTimeColumn = columnMap.get("create_time");
                    // 打印当前批次总共处理的数据行数及最后一行数据的信息
                    if (j == rowDataList.size() - 1 && i == size - 1) {
                        log.info("batchSize = {}, id = {}, createTime = {}", batchSize, this.getColumnValue(idColum), this.getColumnValue(createTimeColumn));
                    }
                }
            }
        } catch (Exception e) {
            log.info("canal message process error: ", e);
        }
    }

    public String getColumnValue(CanalEntry.Column column) {
        return column == null ? null : column.getValue();
    }
}
